package com.atguigu.ct.consumer.dao;

import com.atguigu.ct.common.bean.BaseDao;
import com.atguigu.ct.common.constant.Names;
import com.atguigu.ct.common.constant.ValueConstant;
import com.atguigu.ct.consumer.bean.Calllog;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.List;

/**
 * Description: Hbase数据访问对象
 * @Author: yyx
 * @Create: 2023/4/14 16:27
 * @Version: 8
 */
public class HBaseDao extends BaseDao {
    /**
     * 初始化
     */
    public void init() throws Exception{
        start();

        createNamespaceNX(Names.NAMESPACE.getValue());//有则什么都不做
        createTableXX(Names.TABLE.getValue(),"com.atguigu.ct.consumer.coprocessor.InsertCalleeCoprocessor",ValueConstant.REGION_COUNT,Names.CF_CALLER.getValue(),Names.CF_CALLEE.getValue());//有则覆盖掉

        end();
    }

    /**
     * 插入对象
     * @param calllog
     * @throws Exception
     */
    public void insertData(Calllog calllog) throws Exception{
        calllog.setRowkey(genRegionNum(calllog.getCall1(),
                calllog.getCalltime()) + "_" + calllog.getCall1() + "_" + calllog.getCalltime() +
                        "_" + calllog.getCall2() + "_" + calllog.getDuration());
        putData(calllog);
    }

    /**
     * 插入数据
     * @param value
     */
    public void insertData(String value) throws Exception {
        // 将通话日志保存到Hbase表中
        // 1.获取通话日志数据
        String[] values = value.split("\t");
        String call1 = values[0];
        String call2 = values[1];
        String calltime = values[2];
        String duration = values[3];
        // 2.创建数据对象
            // rowkey设计
            // 1) 长度原则：最大值64KB，推荐长度10~100byte；最好是8的倍数，能短则短，rowkey太长会影响性能
            // 2) 唯一原则
            // 3) 散列原则：
            //   3-1)盐值散列：不能使用时间戳直接作为rowkey；在rowkey前增加随机数
            //   3-2)字符串反转：前面有规律，后面没规律（时间戳、电话号码）
            //   3-3)计算分区号：hashMap
            // rowkey = regionNum + call1 + calltime + call2 + duration
        // 主叫用户
        String rowkey = genRegionNum(call1,calltime) + "_" + call1 + "_" + calltime + "_" + call2 + "_" + duration + "_1";
        Put put = new Put(Bytes.toBytes(rowkey));
        byte[] family = Bytes.toBytes(Names.CF_CALLER.getValue());
        put.addColumn(family,Bytes.toBytes("call1"),Bytes.toBytes(call1));
        put.addColumn(family,Bytes.toBytes("call2"),Bytes.toBytes(call2));
        put.addColumn(family,Bytes.toBytes("calltime"),Bytes.toBytes(calltime));
        put.addColumn(family,Bytes.toBytes("duration"),Bytes.toBytes(duration));
        put.addColumn(family,Bytes.toBytes("flg"),Bytes.toBytes("1"));
        /*String calleeRowkey = genRegionNum(call2,calltime) + "_" + call2 + "_" + calltime + "_" + call1 + "_" + duration + "_0";
        Put callePut = new Put(Bytes.toBytes(calleeRowkey));
        byte[] calleeFamily = Bytes.toBytes(Names.CF_CALLEE.getValue());
        callePut.addColumn(calleeFamily,Bytes.toBytes("call1"),Bytes.toBytes(call2));
        callePut.addColumn(calleeFamily,Bytes.toBytes("call2"),Bytes.toBytes(call1));
        callePut.addColumn(calleeFamily,Bytes.toBytes("calltime"),Bytes.toBytes(calltime));
        callePut.addColumn(calleeFamily,Bytes.toBytes("duration"),Bytes.toBytes(duration));
        callePut.addColumn(calleeFamily,Bytes.toBytes("flg"),Bytes.toBytes("0"));*/
        // 3.保存数据
        List<Put> puts = new ArrayList<Put>();
        puts.add(put);
        //puts.add(callePut);
        putData(Names.TABLE.getValue(),puts);
    }
}
